package SparkRDD.RDD算子.Transformations.聚合操作

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.junit.Test

/*
  * combineByKey 是 reduceByKey和groupByKey 的底层，可控性高，对数据集按照key来进行聚合
  *  createCombiner --- 将values初步进行转换
  *  mergeValue     --- 将初步转换的结果聚合
  *  mergeCombiners --- 在所有分区上把每个分区的聚合结果聚合
  *  partitioner    --- 可选，分区函数
  *  mapSideCombiner--- 可选，是否在map端Combine
  *  serializer     --- 序列化器
  *
 */

class combineByKeyTest {

  val conf = new SparkConf().setMaster("local[6]").setAppName("sample")
  val sc   = new SparkContext(conf)

  @Test
  def combineByKeyTest: Unit ={

    val rdd: RDD[(String, Double)] = sc.parallelize(
      Seq(("张三",98.0),("李四",97.0),("王五",87.0),
      ("张三",94.0),("李四",94.0),("王五",89.0),
      ("李四",94.0),("王五",88.0))
    )

    // 1.createCombiner 转换数据的函数（初始函数，只作用在第一条数据，用于开启整个计算）
    // 2. mergeValue    在分区上的进行聚合
    // 3.mergeCombiners 把所有分区的聚合结果再次聚合成最终结果
    val sumScore = rdd.combineByKey(
      createCombiner = (curr:Double) => (curr,1),
      mergeValue =  (curr:(Double,Int),nextValue:Double) => (curr._1 + nextValue,curr._2 + 1),
      mergeCombiners = (curr:(Double,Int),agg:(Double,Int)) => (curr._1+agg._1,curr._2+agg._2)
    )
    // ("张三“,(98.0+94.0,2))
    // (姓名，(总分，科目数））
    // 计算平均分
    val meanScore = sumScore.map( item => (item._1,item._2._1/item._2._2))

    // 获取结果
    meanScore.collect().foreach( item => println(item))

  }
}
